package org.bdware.doip.core.codec.packet;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.HashedWheelTimer;
import org.apache.log4j.Logger;
import org.bdware.doip.core.doipMessage.MessageEnvelope;
import org.bdware.doip.core.exception.EnvelopeResendException;
import org.bdware.doip.core.exception.MessageCodecException;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.TreeSet;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class GreedyUDPEnvelopeReceiveBuffer extends AbstractEnvelopeReceiveBuffer {

    Logger logger = Logger.getLogger(GreedyUDPEnvelopeReceiveBuffer.class);
    HashMap<Integer,MessageEnvelope> envMap;
    PacketMessageCodec codec = new PacketMessageCodecImpl();
    TreeSet<Integer> sortedPacketNumberSet;
    InetSocketAddress sender;
    ChannelHandlerContext ctx;
    int totalNumber;
    int ackIndex;
    int resendCount = 0; //重传次数

    boolean isTimerTaskRunning = false;
    public static final HashedWheelTimer timer =
            new HashedWheelTimer(Executors.defaultThreadFactory(), 5, TimeUnit.MILLISECONDS, 2);

    public GreedyUDPEnvelopeReceiveBuffer(int requestId, ChannelHandlerContext ctx) {
        super(requestId);
        envMap = new HashMap<>();
        sortedPacketNumberSet = new TreeSet<>();
        this.ctx = ctx;
        totalNumber = 0;
        ackIndex = 0;
    }

    @Override
    public boolean isCompleted() {
        return (totalNumber == (ackIndex+1));
    }

    @Override
    public int addEnvelope(MessageEnvelope envelope) throws EnvelopeResendException {
        if(totalNumber == 0 ) totalNumber = envelope.totalNumber;
        else if(totalNumber!=envelope.totalNumber)
            throw new EnvelopeResendException("invalid total number, expected: " + totalNumber + " got: " + envelope.totalNumber);
        envMap.put(envelope.sequenceNumber,envelope);
        sortedPacketNumberSet.add(envelope.sequenceNumber);
        if(sender == null) sender = envelope.sender;
        resend();
        return totalNumber - envMap.size();
    }

    @Override
    public ArrayList<MessageEnvelope> getSortedEnvelopes() {
        //升序排列Envelope
        ArrayList<MessageEnvelope> envelopes = new ArrayList<>();
        for(int key:sortedPacketNumberSet){
           envelopes.add(envMap.get(key));
        }
        return envelopes;
    }

    /**
     * 两个重传策略：发现当前接收包序列中有丢包，则重传丢包
     * 若没有丢包且未完成，则500ms之后请求重传当前ack + 100的包
     */
    public void resend(){
        if (resendCount >= 1000) {
            logger.warn("resend over 1000 times, abandon...");
            return;
        }
        TreeSet<Integer> lostPackets = getLostPacketNumber();
        if(isCompleted()){
            MessageEnvelope resendMsg = MessageEnvelope.createResendMessage(requestId,-1,sender);

//            logger.info("try to send complete ack, requestID: " + requestId);
//            logger.info("try to send complete ack, sender: " + resendMsg.getSender().toString());
//            logger.info("resend flag: " + resendMsg.isResend());
            try {
                sendResendMessage(resendMsg);
            } catch (MessageCodecException e) {
                e.printStackTrace();
            }
            return;
        }
        for (int seqNumber:lostPackets) {
            MessageEnvelope resendMsg = MessageEnvelope.createResendMessage(requestId,seqNumber,sender);
            logger.info("resendCount: " + resendCount + "; sequence number: "+ seqNumber);
            ctx.writeAndFlush(resendMsg);
            resendCount++;
        }
        if(lostPackets.isEmpty() && !isCompleted() && !isTimerTaskRunning){
            logger.debug("ackIndex: " + ackIndex);
            isTimerTaskRunning = true;
            timer.newTimeout(timeout -> {
                if(isCompleted()) return;
                MessageEnvelope resendMsg = MessageEnvelope.createResendMessage(
                        requestId,
                        ackIndex<(totalNumber-100)?totalNumber:ackIndex+100,
                        sender);
                isTimerTaskRunning = false;
                logger.info("resendCount: " + resendCount + "; sequence number: "+ (ackIndex<(totalNumber-100)?totalNumber:ackIndex+100));
                sendResendMessage(resendMsg);
                resendCount++;
            },
                    500,
                    TimeUnit.MILLISECONDS);
        }
    }

    TreeSet<Integer> getLostPacketNumber(){
        TreeSet<Integer> lostPackets = new TreeSet<>();
        for(int i=ackIndex+1; i <= sortedPacketNumberSet.last(); i++){
            if(envMap.get(i) == null){
                lostPackets.add(i);
            }else if(lostPackets.isEmpty()){
                ackIndex = i;
            }
        }
        return lostPackets;
    }

    public void sendResendMessage(MessageEnvelope env) throws MessageCodecException {
        ByteBuf buff = Unpooled.directBuffer();
        buff.writeBytes(codec.EnvelopeToBytes(env));
        ctx.writeAndFlush(new DatagramPacket(buff,env.getSender()));
    }

}
